package cloud.xiguapi.ubas.analysis.login;

import cloud.xiguapi.ubas.analysis.login.model.LoginEvent;
import cloud.xiguapi.ubas.analysis.login.model.LoginFailWarning;
import cloud.xiguapi.ubas.analysis.login.model.TimelinessLoginFailDetectWarning;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

import static cloud.xiguapi.ubas.constant.JobNameConstant.LOGIN_FAIL;

/**
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-19 下午 06:13
 * desc:
 */
@Component
public class TimelinessLoginFailAnalysis {

    @Resource(name = "env")
    private StreamExecutionEnvironment env;

    @Async("timelinessLoginFailAnalysis")
    public void analysis(String path) throws Exception {
        env.setParallelism(8);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 读取数据并转换
        SingleOutputStreamOperator<LoginEvent> loginEventDataStream = env.readTextFile(path)
                .map(line -> {
                    String[] fields = line.split(",");
                    return LoginEvent.builder()
                            .userId(Long.valueOf(fields[0]))
                            .ip(fields[1])
                            .loginState(fields[2])
                            .timestamp(Long.valueOf(fields[3]))
                            .build();
                })
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LoginEvent>(Time.seconds(3)) {
                    @Override
                    public long extractTimestamp(LoginEvent element) {
                        return element.getTimestamp() * 1000L;
                    }
                });

        SingleOutputStreamOperator<LoginFailWarning> loginFailDataStream = loginEventDataStream
                .keyBy(LoginEvent::getUserId)
                .process(new TimelinessLoginFailDetectWarning());

        loginFailDataStream.print();
        loginFailDataStream.addSink(new TimelinessLoginFailMySqlSink());

        env.execute(LOGIN_FAIL.getJobName());
    }

    public static class TimelinessLoginFailMySqlSink extends RichSinkFunction<LoginFailWarning> {

        private transient Connection connection;

        private transient PreparedStatement clearStatement;

        private transient PreparedStatement insertStatement;

        private transient boolean isFirst = true;

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/ubas?serverTimezone=UTC", "root", "1015");
            clearStatement = connection.prepareStatement("TRUNCATE TABLE login_fail");
            insertStatement = connection.prepareStatement("INSERT INTO login_fail(user_id, first_fail, last_fail, warning) VALUES(?, ?, ?, ?)");
        }

        @Override
        public void close() throws Exception {
            clearStatement.close();
            insertStatement.close();
            connection.close();
        }

        @Override
        public void invoke(LoginFailWarning value, SinkFunction.Context context) throws Exception {

            if (isFirst) {
                clearStatement.execute();
                isFirst = false;
            }

            insertStatement.setLong(1, value.getUserId());
            insertStatement.setLong(2, value.getFirstFail());
            insertStatement.setLong(3, value.getLastFail());
            insertStatement.setString(4, value.getWarning());

            insertStatement.execute();
        }
    }
}
